package cn.itcast.streaming.function.watermark;

import cn.itcast.streaming.bean.ElectricFenceModel;
import cn.itcast.streaming.bean.ItcastDataObj;
import cn.itcast.streaming.utils.DateUtil;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;

//电子围栏水印操作
public class ElectricFenceWaterMark implements AssignerWithPeriodicWatermarks<ElectricFenceModel>{

    //当前窗口最大水印时间
    private long currentMaxTimestamp = 0L;
    //允许最大的乱序时间30S
    private final long maxOutofOrderness = 30*1000L;

    @Nullable
    @Override //生成当前水印
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxTimestamp - maxOutofOrderness);
    }

    @Override
    public long extractTimestamp(ElectricFenceModel electricFenceModel, long l) {
        currentMaxTimestamp = Math.max(electricFenceModel.getTerminalTimestamp(),currentMaxTimestamp);
        System.out.println("当前数据时间："+ DateUtil.timestampToDataStr(electricFenceModel.getTerminalTimestamp())+" 窗口最大时间："+DateUtil.timestampToDataStr(currentMaxTimestamp));
        return currentMaxTimestamp;
    }
}
